package com.hitqz.robot.biz.websocket.endpoint;

import cn.hutool.core.collection.CollUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.HashBasedTable;
import com.google.common.collect.Table;
import com.hitqz.robot.api.common.enums.SubscribeTopicEnum;
import com.hitqz.robot.biz.config.LogTailer;
import com.hitqz.robot.biz.websocket.config.CustomConfigurator;
import jakarta.annotation.PreDestroy;
import jakarta.websocket.*;
import jakarta.websocket.server.PathParam;
import jakarta.websocket.server.ServerEndpoint;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Arrays;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;

@ServerEndpoint(value = "/subscribe/{topic}/{randomId}",configurator = CustomConfigurator.class)
@Component
@Slf4j
public class SubscriptionEquipmentStatusEndpoint {

	private static final ObjectMapper objectMapper = new ObjectMapper();
	private static final Table<String, Integer, CopyOnWriteArraySet<Session>> topicSubscribers = HashBasedTable.create();
	private static final Set<Session> sessions = new CopyOnWriteArraySet<>();

	@OnOpen
	public void onOpen(Session session,
					   @PathParam(value = "topic") String topic,
					   @PathParam(value = "randomId") String randomId) throws IOException {
		try {
			SubscribeTopicEnum subscribeTopicEnum = Arrays.stream(SubscribeTopicEnum.values()).filter(
					t-> t.getTopic().equals(topic)).findFirst().get();
			sessions.add(session);
			subscribe(session, subscribeTopicEnum.getTopic(), Integer.parseInt(randomId));
			if (topic.equals(SubscribeTopicEnum.ROBOT_PUSH_INFO_LOG.getTopic())) {
				LogTailer.pushNow(null,randomId);
			}
		}catch (Exception e) {
			if (log.isWarnEnabled()) {
				session.getBasicRemote().sendText("无法解析该话题信息");
				log.warn("话题关注发生异常无法解析 {}", randomId);
			}
		}
	}

	@OnMessage
	public void onMessage(Session session,String message) {
		log.info("[ws][subscribe],message:{},session-id:{}", message,session.getId());
		// 如果是设备的话需要关注


	}

	@OnClose
	public void onClose(Session session,@PathParam(value = "topic") String topic,
						@PathParam(value = "randomId") String randomId) {
		//取消订阅
		unsubscribe(session, Integer.parseInt(randomId));
		// 移除session
		sessions.remove(session);
		if (log.isDebugEnabled()) {
			log.debug("[ws][subscribe],disconnected.current size :{},session-id:{}",sessions.size(),session.getId());
		}
	}

	@OnError
	public void onError(Session session,Throwable error) {
		log.error("[ws][subscribe],error:{},session-id:{}",error.getMessage(),session.getId());
	}

	// 话题广播
	public static void broadcast(Integer randomId, SubscribeTopicEnum subscribeTopicEnum, String message) {
		if (randomId == null) {
			Map<Integer, CopyOnWriteArraySet<Session>> setMap = topicSubscribers.row(subscribeTopicEnum.getTopic());
			for (Integer key : setMap.keySet()) {
				SubscriptionEquipmentStatusEndpoint.broadcast(key, subscribeTopicEnum, message);
			}
		} else {
			CopyOnWriteArraySet<Session> sessionCopyOnWriteArraySet = topicSubscribers.get(subscribeTopicEnum.getTopic(),randomId);
			if (CollUtil.isNotEmpty(sessionCopyOnWriteArraySet)) {
				for (Session subscriber : sessionCopyOnWriteArraySet) {
					if (subscriber.isOpen()) {
						subscriber.getAsyncRemote().sendText(message);
					}
				}
			}
		}
	}

	// 订阅话题
	public static void subscribe(Session session, String topic, Integer equipmentId) {
		CopyOnWriteArraySet<Session> sets = topicSubscribers.get(topic, equipmentId);
		if (CollUtil.isEmpty(sets)) {
			sets = new CopyOnWriteArraySet<>();
		}
		sets.add(session);
		topicSubscribers.put(topic, equipmentId, sets);
		log.info("[ws][subscribe],subscribe topic:{},session-id:{}",topic,session.getId());
	}

	/**
	 * 取消订阅
 	 */
	public static void unsubscribe(Session session, Integer equipmentId) {
		if (session == null || equipmentId == null) {
			throw new IllegalArgumentException("Session and equipmentId cannot be null");
		}

		// 移除订阅session
		Map<String, CopyOnWriteArraySet<Session>> setMap = topicSubscribers.column(equipmentId);
		if (setMap != null) {
			for (String key : setMap.keySet()) {
				CopyOnWriteArraySet<Session> keySession = setMap.get(key);
				if (keySession != null) {
					keySession.remove(session);
					setMap.put(key, keySession);
				}
			}
		}

		log.info("[ws][subscribe],unsubscribe topic:%s,session-id:%s", equipmentId, session.getId());
	}

	private void sendError(Session session, String errorMessage) {
		try {
			session.getBasicRemote().sendText("{\"error\":\"" + errorMessage + "\"}");
		} catch (IOException e) {
			// 处理发送错误消息时的异常
		}finally {
			try {
				// 在发送错误信息后关闭会话
				session.close(new CloseReason(CloseReason.CloseCodes.CANNOT_ACCEPT, errorMessage));
			} catch (IOException e) {
				// 处理关闭session时的异常
				if (log.isErrorEnabled()) {
					log.error("[ws][subscribe]close, session-id:{} failed:{}",session.getId(),e.getMessage());
				}
			}
		}
	}

    @PreDestroy
    public void onDestroy() {
        log.info("Closing all WebSocket connections");
        for (Session session : sessions) {
            try {
                if (session.isOpen()) {
                    session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "Application is shutting down"));
                }
            } catch (IOException e) {
                log.error("Error closing WebSocket session: " + session.getId(), e);
            }
        }
    }


}
